package coml.zhouxiaoge.zookeeper;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author zhouxiaoge
 * <p>
 * Created by 2022-04-27 21:23
 */
public class DistributedLock {

    private static final String connectString = "192.168.1.11:2181,192.168.1.12:2181,192.168.1.13:2181";
    private static final int sessionTimeout = 2000;
    private final static String root = "/locks";
    private final static String subNode = "/seq-";

    private final ZooKeeper zooKeeper;
    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);

    private String waitPath;
    private String currentMode;

    public DistributedLock() throws Exception {
        // 获取连接
        zooKeeper = new ZooKeeper(connectString, sessionTimeout, event -> {
            // connectLatch 如果连接上zookeeper，可以释放
            if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
                connectLatch.countDown();
            }
            // waitLatch 需要释放
            if (event.getType() == Watcher.Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                waitLatch.countDown();
            }
        });
        // 等待zookeeper正常连接后，再往下走
        connectLatch.await();
        // 判断根节点/locks是否存在
        Stat exists = zooKeeper.exists(root, false);
        if (exists == null) {
            //创建一下根节点
            zooKeeper.create(root, "locks".getBytes(StandardCharsets.UTF_8), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

    }

    // 对zookeeper加锁
    public void zkLock() throws Exception {
        // 创建临时带序号节点
        currentMode = zooKeeper.create(root + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        // 判断创建的节点是否是最小序号节点，如果是获取到锁，如果不是，监听它前一个节点
        List<String> children = zooKeeper.getChildren(root, false);
        if (children.size() == 1) {
            return;
        } else {
            Collections.sort(children);
            String thisNode = currentMode.substring((root + "/").length());
            int index = children.indexOf(thisNode);
            if (index == -1) {
                System.out.println("数据异常");
            } else if (index == 0) {
                return;
            } else {
                // 需要监听，它前一个节点变化
                waitPath = root + "/" + children.get(index - 1);
                zooKeeper.getData(waitPath, true, null);
                // 等待监听
                waitLatch.await();
                return;
            }
        }

    }

    // 解锁
    public void unZkLock() throws InterruptedException, KeeperException {
        // 删除节点
        zooKeeper.delete(currentMode, -1);
    }
}
